/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.activity.impl;

import java.util.concurrent.atomic.AtomicInteger;

import org.sabayframework.activity.ActivityContext;
import org.sabayframework.channel.Lock;
import org.sabayframework.channel.ram.ArrayLock;


/**
 * Fork replicates and executes given activity for each input object.
 * 
 * @author Alex Antonau
 * */
public class Fork<I, O> extends StatefulActivity<I, O> {

	private final ReplicableActivity<I, O> activity;
	private final AtomicInteger counter;
	private int maxActivities;
	private boolean started;
	
	public Fork(ReplicableActivity<I, O> activity) {
		this(activity, 0);
	}

	/**
	 * @param activity replicable activity
	 * @param maxActivities maximum number of concurrent activities
	 * */
	public Fork(ReplicableActivity<I, O> activity, int maxActivities) {
		this.activity = activity;
		this.counter = new AtomicInteger();
		this.maxActivities=maxActivities;
	}

	@Override
	public void beforeExecute(ActivityContext<I, O> ctx) throws Exception {
		if(maxActivities <=0) maxActivities = ctx.getExecutionQueue().workerCount();
		activity.beforeExecute(ctx);
	}

	@Override
	public void afterExecute(ActivityContext<I, O> ctx) throws Exception {
		activity.afterExecute(ctx);
	}

	@Override
	public boolean isReady(ActivityContext<I, O> ctx) throws Exception {
		return activity.isReady(ctx) && counter.get() < maxActivities;
	}

	@Override
	public boolean isDone(ActivityContext<I, O> ctx) throws Exception {
		return activity.isDone(ctx) && counter.get() == 0;
	}

	@Override
	public int getOutputSize(ActivityContext<I, O> ctx) {
		return 0;
	}

	@Override
	public int getInputSize(ActivityContext<I, O> ctx) {
		return activity.getInputSize(ctx);
	}
	final class ForkedActivity extends StatefulActivity<I,O>{
		
		private final Lock<I> inputLock;
		private final ReplicableActivity<I, O> replica;
		private boolean done;
		
		public ForkedActivity(ReplicableActivity<I, O> replica, Lock<I> inputLock) {
			super();
			this.replica = replica;
			this.inputLock = new ArrayLock<I>(inputLock.toArray());
		}

		@Override
		public void beforeExecute(ActivityContext<I, O> ctx) {}

		@Override
		public void afterExecute(ActivityContext<I, O> ctx) {
			counter.decrementAndGet();
		}

		@Override
		public boolean isReady(ActivityContext<I,O> ctx) throws Exception {
			if(done) return false;
			if(getState(ctx)==State.READY) return true;
			return super.isReady(ctx);
		}

		@Override
		public boolean isDone(ActivityContext<I,O> ctx) throws Exception {
			return done || super.isDone(ctx);
		}

		@Override
		public void execute(ActivityContext<I, O> ctx) throws Exception {
			if(getState(ctx)==State.READY){
				replica.input(ctx, inputLock);
				setState(ctx, State.PROCESS);
			}
			super.execute(ctx);
		}

		@Override
		public int getOutputSize(ActivityContext<I, O> ctx) {
			return replica.getOutputSize(ctx);
		}

		@Override
		public void input(ActivityContext<I, O> ctx, Lock<I> inputLock) throws Exception {
			throw new RuntimeException("Not implemented");
		}

		@Override
		public void process(ActivityContext<I, O> ctx) throws Exception {
			replica.process(ctx);
		}

		@Override
		public void output(ActivityContext<I, O> ctx, Lock<O> outputLock) throws Exception {
			replica.output(ctx, outputLock);
			done = true;
		}
		
	}

	@Override
	public void execute(ActivityContext<I, O> ctx) throws Exception {
		final int count = counter.incrementAndGet();
		started = false;
		try{
			if(count <= maxActivities) {
				super.execute(ctx);
			}
		} finally {
			//decrement only if no forked activity submitted
			if(!started) counter.decrementAndGet();
		}
	}

	@Override
	public void input(ActivityContext<I, O> ctx, Lock<I> inputLock) throws Exception {
		ReplicableActivity<I, O> replica = activity.replicate();
		ctx.getExecutionQueue().submit(new ForkedActivity(replica, inputLock), ctx.getInput(), ctx.getOutput());
		started = true;
	}

	@Override
	public void output(ActivityContext<I, O> ctx, Lock<O> outputLock)throws Exception {
	}

}
